package com.ikingtech.framework.sdk.cluster.node.handler;

import com.ikingtech.framework.sdk.cluster.client.ClusterClient;
import com.ikingtech.framework.sdk.cluster.node.ClusterNode;
import com.ikingtech.framework.sdk.cluster.node.ClusterNodeManager;
import com.ikingtech.framework.sdk.utils.Tools;
import io.netty.buffer.ByteBuf;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;

/**
 * @author tie yan
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ClusterNodeJoinMessageHandler implements ClusterNodeMessageHandler {

    private final ClusterNodeManager nodeManager;

    private final ClusterClient client;

    @Override
    public void handle(Object message) {
        ClusterNodeJoinMessage nodeJoinMessage = Tools.Json.toBean(((ByteBuf) message).toString(Charset.defaultCharset()), ClusterNodeJoinMessage.class);
        if (null == nodeJoinMessage) {
            log.warn("[CLUSTER]node join message is null");
            return;
        }
        ClusterNode node = this.nodeManager.get(nodeJoinMessage.getAddress(), nodeJoinMessage.getPort());
        if (null != node) {
            log.debug("[CLUSTER][NODE-JOIN]node(address={}, port={}) has been joined", nodeJoinMessage.getNode().getAddress(), nodeJoinMessage.getNode().getPort());
            return;
        }
        log.debug("[CLUSTER][NODE-JOIN]new node(address={}, port={}) join, put into node list", nodeJoinMessage.getNode().getAddress(), nodeJoinMessage.getNode().getPort());
        this.client.newConnection(nodeJoinMessage.getNode());
        this.nodeManager.put(nodeJoinMessage.getNode());
    }
}
